package net.bwie.flink;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Flink CDC从MySQL表中实时获取数据，并解析为json数据
 * @author xuanyu
 * @date 2025/10/16
 */
public class _04FlinkCdcMySQLDemo {

	public static void main(String[] args) throws Exception{
		// 1.执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment() ;
		env.setParallelism(1) ;
		env.enableCheckpointing(3000);

		// 2. 数据源-source
		MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
			.hostname("node101")
			.port(3306)
			.databaseList("db_test")
			.tableList("db_test.tbl_users")
			.username("root")
			.password("123456")
			.startupOptions(StartupOptions.initial())
			.deserializer(new JsonDebeziumDeserializationSchema())
			.build();
		DataStreamSource<String> stream = env.fromSource(
			mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source"
		);

		// 3. 数据转换-transformation

		// 4. 数据接收器-sink
		stream.print();

		// 5. 触发器-execute
		env.execute("FlinkCdcMySQLDemo") ;
	}

}
